package com.patsnap.data.npd.dw.core.pipeline.journal;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSONObject;
import com.patsnap.data.npd.dw.core.pipeline.AbstractPipeline;
import com.patsnap.data.npd.dw.core.pipeline.PipelineContext;
import com.patsnap.data.npd.dw.core.pipeline.PipelineException;
import com.patsnap.data.npd.dw.repository.DBException;
import com.patsnap.data.npd.dw.repository.DBOperator;
import com.patsnap.data.npd.dw.repository.impl.AdsJournalRepositoryImpl;
import com.patsnap.data.npd.dw.schema.ads.journal.AdsJournal;
import com.patsnap.data.npd.dw.schema.dwd.journal.DwdJournalJBestInfo;
import com.patsnap.data.npd.dw.schema.ods.journal.OdsJournalJBestInfo;
import com.patsnap.one.etl.cdc.CdcRecord;
import lombok.Data;
import lombok.experimental.SuperBuilder;

public class DwdJournalCalculationPipeline extends AbstractPipeline<DwdJournalCalculationPipeline.DwdJournalCalculationContext> {

    @Override
    public String pipelineName() {
        return "Journal data to dwd job";
    }

    @Override
    public DwdJournalCalculationContext initContext(CdcRecord cdcRecord) {
        return DwdJournalCalculationPipeline.DwdJournalCalculationContext.builder().cdcRecord(cdcRecord).build();
    }

    @Override
    public void extract(DBOperator client, DwdJournalCalculationContext context) throws PipelineException {
        CdcRecord cdcRecord = context.getCdcRecord();
        String table = cdcRecord.getSource().getTable();
        JSONObject dataBody = null != cdcRecord.getAfter() ? cdcRecord.getAfter() : cdcRecord.getBefore();
        switch (table) {
            case "ods_journal_jbest_info":
                context.setOdsJournalJBestInfo(dataBody.toJavaObject(OdsJournalJBestInfo.class));
                break;
            default:
                throw new PipelineException("Unsupported table: " + table);
        }
    }

    @Override
    public void transform(DBOperator client, DwdJournalCalculationContext context) throws PipelineException {

    }

    @Override
    public void load(DBOperator client, DwdJournalCalculationContext context) throws PipelineException {
        try {
            OdsJournalJBestInfo odsJournalJBestInfo = context.getOdsJournalJBestInfo();
            if (odsJournalJBestInfo != null) {
                DwdJournalJBestInfo dwdJournalJBestInfo = new DwdJournalJBestInfo();
                BeanUtil.copyProperties(odsJournalJBestInfo, dwdJournalJBestInfo);

                // 获取 journal_id, 依次通过 issn 和 eissn 查询 journal_id
                AdsJournalRepositoryImpl adsJournalRepository = client.getAdsJournalRepository();
                AdsJournal adsJournal = adsJournalRepository.queryByIssn(odsJournalJBestInfo.getIssn());
                if (adsJournal != null) {
                    dwdJournalJBestInfo.setJournalId(adsJournal.getJournalId());
                }

                client.getDwdJournalRepository().upsert(dwdJournalJBestInfo, DwdJournalJBestInfo.class);
            }
        } catch (DBException e) {
            throw new PipelineException(e);
        }
    }

    @Data
    @SuperBuilder
    public static class DwdJournalCalculationContext extends PipelineContext {
        private OdsJournalJBestInfo odsJournalJBestInfo;
    }
}
